package sdm.sensor.nasdaq.floe;

import java.util.Date;
import java.text.DateFormat;
import java.text.SimpleDateFormat;

import java.io.*;

import org.apache.log4j.*;

import com.espertech.esper.client.*;



// http://coffeeonesugar.wordpress.com/2009/07/21/getting-started-with-esper-in-5-minutes/

public class Nasdaq {

	public static void main(String args[]) {
		
		
		boolean redirect = false;
		// redirect 'System.out' to file
		if (redirect) redirectSystemOut();
			
		// print timestamp
		DateFormat dateFormat = new SimpleDateFormat("yyyy.MM.dd HH:mm:ss");
        System.out.println(dateFormat.format(new Date()).toString());
		
		// init log4j
		SimpleLayout layout = new SimpleLayout();
		ConsoleAppender appender = new ConsoleAppender(layout);
		Logger.getRootLogger().addAppender(appender);
		Logger.getRootLogger().setLevel((Level) Level.WARN);
		
		
		// start esper
		
		//The Configuration is meant only as an initialization-time object.
		Configuration cepConfig = new Configuration();
	    // We register Ticks as objects the engine will have to handle
	    cepConfig.addEventType("Tick",Tick.class.getName());
	    // We setup the engine
	    EPServiceProvider cepService = EPServiceProviderManager.getProvider("myCEPEngine",cepConfig);
		
	    EPRuntime cepRuntime = cepService.getEPRuntime();

		
		// get Stocks
//	    Stock stockApple = new Stock("AAPL", redirect);
//	    Stock stockGoogle = new Stock("GOOG", redirect);
	    Stock stockMs = new Stock("MSFT", redirect);
		Stock stockApple = new Stock("ATX", redirect);
	    Stock stockGoogle = new Stock("EURUSD=X", redirect);
	    
	    // 1.) Statement 'average' --> get the average of a 20sec window after every event
		String avgExpression = 
				"select *, avg(price) as avg " +
				"from Tick.std:groupwin(symbol).win:time(20 sec) " +	// --> for counter-Test-Data set to length(20)
				"group by symbol";
		EPStatement avgStatement = 
				cepService.getEPAdministrator().createEPL(avgExpression);
				
		// Listener 'average'
		class avgCEPListener implements UpdateListener { // Interface UpdateListener (inner class)
			public void update(EventBean[] newEvents, EventBean[] oldEvents) {
				for (EventBean event : newEvents) {
					//EventBean event = newEvents[0];
					System.out.println("(1) avg (" + event.get("symbol") + "): " + event.get("avg"));
					//System.out.println("Event received: " + event.getUnderlying());
				}
			}
		};
		
		avgStatement.addListener(new avgCEPListener());
		
		
		// 2.) Statement 'max' --> get the max of a 20sec window after every event
		String maxExpression = 
				"select *, max(price) as max " +
				"from Tick.std:groupwin(symbol).win:time(20 sec) " +	// --> for counter-Test-Data set to length(20)
				"group by symbol";
		EPStatement maxStatement = 
				cepService.getEPAdministrator().createEPL(maxExpression);
				
		// Listener 'max'
		class maxCEPListener implements UpdateListener {
			public void update(EventBean[] newEvents, EventBean[] oldEvents) {
				for (EventBean event : newEvents) {
					System.out.println("(2) max (" + event.get("symbol") + "): " + event.get("max"));
					//System.out.println("Event received: " + event.getUnderlying());
				}
			}
		};
		
		maxStatement.addListener(new maxCEPListener());		
		
		
		// 3.) Statement 'deltaAvg' --> get the difference of 
		//     the avg of two following 20sec windows
		
		// load stream
		String deltaAvgCreateExpression = 
				"insert into deltaAvgTicks " +
				"select symbol, " +
				"avg(price) as avgPrice " +
				"from Tick.win:time_batch(20 sec) " +	// --> for counter-Test-Data set to length_batch(20)
				"group by symbol";
		@SuppressWarnings("unused")
		EPStatement deltaAvgCreateStatement = 
				cepService.getEPAdministrator().createEPL(deltaAvgCreateExpression);
				
		// calculate delta
		String deltaAvgExpression = 
			"select symbol, " +
			"avgPrice, " +
			"prior(3, avgPrice) as avgPricePrior, " + 							// --> 3 = count(symbol)
			"(avgPrice-prior(3, avgPrice))/prior(3, avgPrice) as deltaAvg " +	// --> 3 = count(symbol)
			"from deltaAvgTicks.win:length(1) " +
			"group by symbol";
		EPStatement deltaAvgStatement = 
			cepService.getEPAdministrator().createEPL(deltaAvgExpression);
			

		// Listener 'deltaMax'
		class deltaAvgCEPListener implements UpdateListener {
			public void update(EventBean[] newEvents, EventBean[] oldEvents) {
				for (EventBean event : newEvents) {
					System.out.println("(3) deltaAvg (" + event.get("symbol") + "): " + event.get("deltaAvg"));
					//System.out.println("Event received: " + event.getUnderlying());
				}
				
			}
		};
		
		deltaAvgStatement.addListener(new deltaAvgCEPListener());	
		
		
		
		
		// 4.) Statement 'deltaMax' --> get the difference of 
		//     the Max of two following 20sec windows
		
		// load stream
		String deltaMaxCreateExpression = 
				"insert into deltaMaxTicks " +
				"select symbol, " +
				"max(price) as maxPrice " +
				"from Tick.win:time_batch(20 sec) " + 	// --> for counter-Test-Data set to length_batch(20)
				"group by symbol";
		@SuppressWarnings("unused")
		EPStatement deltaMaxCreateStatement = 
				cepService.getEPAdministrator().createEPL(deltaMaxCreateExpression);
				
		// calculate delta
		String deltaMaxExpression = 
			"select symbol, " +
			"maxPrice, " +
			"prior(3, maxPrice) as maxPricePrior, " + 		// --> 3 = count(symbol)
			"maxPrice - prior(3, maxPrice) as deltaMax " +	// --> 3 = count(symbol)
			"from deltaMaxTicks.win:length(1) " +
			"group by symbol";
		EPStatement deltaMaxStatement = 
			cepService.getEPAdministrator().createEPL(deltaMaxExpression);
			

		// Listener 'deltaMax'
		class deltaMaxCEPListener implements UpdateListener {
			public void update(EventBean[] newEvents, EventBean[] oldEvents) {
				for (EventBean event : newEvents) {
					System.out.println("(4) deltaMax (" + event.get("symbol") + "): " + event.get("deltaMax"));
					System.out.println("Event received: " + event.getUnderlying());
				}
				
			}
		};
		
		deltaMaxStatement.addListener(new deltaMaxCEPListener());	
		
	
    
	    // alle 2sec für 5min --> 150 Samples
	    for (int i=0; i<150; i++) {
	    	// send new Tick from Stock to EPRuntime
	    	System.out.println("\n  SAMPLE " + i + "\t\t\t\t" + dateFormat.format(new Date()).toString());
	    	
	    	// get Stock Data
//	    	stockApple.refreshCurrentTick();
//	    	stockGoogle.refreshCurrentTick();
//	    	stockMs.refreshCurrentTick();
	    	
	    	// Test data - random Stock
//	    	stockApple.randomCurrentTick();
//	    	stockGoogle.randomCurrentTick();
//	    	stockMs.randomCurrentTick();	    	

	    	stockApple.refreshHistoricTick();
	    	stockGoogle.refreshHistoricTick();
	    	stockMs.refreshHistoricTick();

	    	
	    	// Debug
	    	//System.out.println(stockApple);
	    	//System.out.println(stockGoogle);
	    	//System.out.println(stockMs);
	    	
	    	// send Stock Event
	    	cepRuntime.sendEvent(stockApple.getCurrentTick());
	    	cepRuntime.sendEvent(stockGoogle.getCurrentTick());
	    	cepRuntime.sendEvent(stockMs.getCurrentTick());
	    	
	    	// Test data - counter
//	    	cepRuntime.sendEvent(new Tick("AAPL", i));
//	    	cepRuntime.sendEvent(new Tick("GOOG", 50-i));
//	    	cepRuntime.sendEvent(new Tick("MSFT", 10*i));
	    	
	    	try {
				Thread.sleep(2000);// --> for counter-Test-Data set Thread.sleep(0)
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
	    }
	    
	    // close Stocks
	    stockApple.close();
	    stockGoogle.close();
	    stockMs.close();
		
	}
	
	private static void redirectSystemOut() {
		PrintStream defaultStream 	= null;
		PrintStream newStream 	= null;
		
		DateFormat dateFormat = new SimpleDateFormat("yyyy.MM.dd HH:mm:ss");
		
		try {
			// saving the default stream
			defaultStream = System.out;
			newStream = new PrintStream(new FileOutputStream(
					"data/NASDAQ__" + "__" +
					dateFormat.format(new Date()).toString() + ".txt",true));
			// redirecting console output to file
			System.setOut(newStream);
			// redirecting runtime exceptions to file
			System.setErr(newStream);
		}
		catch (FileNotFoundException fnfEx)
		{
			System.out.println("Error in IO Redirection");
			fnfEx.printStackTrace();
			
			// resetting stream (although it shouldn't have changed at this point...)
			System.setOut(defaultStream);
			System.setErr(defaultStream);
		}
		
	}
		
}
